package transform;

import bean.SensorReading;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class TransDemo02_RollingAggregation {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> inputDS = env.readTextFile("D:\\CS\\MyDemo\\gmall-flink\\flink-demo\\src\\main\\resources\\sensor.txt");

        SingleOutputStreamOperator<SensorReading> dataStream = inputDS.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0], new Long(fields[1]), new Double(fields[2]));
        });

        KeyedStream<SensorReading, Tuple> keyedStream = dataStream.keyBy("id");
//        11> SensorReading(id=sensor_6, timestamp=1547718201, temperature=15.4)
//        13> SensorReading(id=sensor_7, timestamp=1547718202, temperature=6.7)
//        8> SensorReading(id=sensor_10, timestamp=1547718205, temperature=38.1)
//        10> SensorReading(id=sensor_1, timestamp=1547718207, temperature=36.3)
//        10> SensorReading(id=sensor_1, timestamp=1547718209, temperature=32.8)
//        10> SensorReading(id=sensor_1, timestamp=1547718199, temperature=35.8)
//        10> SensorReading(id=sensor_1, timestamp=1547718212, temperature=37.1)

        KeyedStream<SensorReading, String> keyedStream1 = dataStream.keyBy(data -> data.getId());

        KeyedStream<SensorReading, String> keyedStream2 = dataStream.keyBy(new KeySelector<SensorReading, String>() {
            @Override
            public String getKey(SensorReading sensorReading) throws Exception {
                return sensorReading.getId();
            }
        });

        SingleOutputStreamOperator<SensorReading> resultStream = keyedStream2.maxBy("temperature");

        resultStream.print("result");
        env.execute();

    }
}
